package com.ynet.eq.mqtt.consumer;

import lombok.extern.slf4j.Slf4j;
import org.eclipse.paho.client.mqttv3.*;
import org.springframework.core.io.support.SpringFactoriesLoader;

import java.util.Arrays;
import java.util.Iterator;
import java.util.List;
import java.util.ServiceLoader;

/**
 * @author 胡超雄
 * @date 2021/6/30 10:23
 * @desc mqtt回调处理类
 */
@Slf4j
public class MqttConsumerCallback implements MqttCallbackExtended {

    private MqttClient client;
    private MqttConnectOptions options;
    private String[] topic;
    private int[] qos;
    private String clientId;


    public MqttConsumerCallback(MqttClient client, MqttConnectOptions options, String[] topic, int[] qos, String clientId) {
        this.client = client;
        this.options = options;
        this.topic = topic;
        this.qos = qos;
        this.clientId = clientId;
    }
    //设置相关属性


    /**
     * mqtt连接后订阅主题
     *
     * @param b
     * @param s
     */
    @Override
    public void connectComplete(boolean b, String s) {
        try {
            if (null != topic && null != qos) {
                if (client.isConnected()) {
                    client.subscribe(topic, qos);
                    log.debug("mqtt连接成功，客户端ID：" + clientId);
                    log.debug("--订阅主题:：" + Arrays.toString(topic));
                } else {
                    log.error("mqtt连接失败，客户端ID" + clientId);
                }
            }
        } catch (Exception e) {
            log.error("mqtt订阅主题异常:" + e);
        }
    }

    /**
     * 断开重连
     *
     * @param throwable
     */
    @Override
    public void connectionLost(Throwable throwable) {
        log.debug("MQTT连接断开，发起重连中......");
        try {
            if (null != client && !client.isConnected()) {
                client.reconnect();
                log.debug("尝试重新连接.");
            } else {
                client.connect(options);
                log.debug("尝试建立新连接");
            }
        } catch (Exception e) {
            log.error("尝试重新连接失败", e);
            e.printStackTrace();
        }
    }

    /**
     * 消息处理
     *
     * @param s       接受到了某个消息 开始处理.... 此处使用spi机制
     * @param message
     */
    @Override
    public void messageArrived(String s, MqttMessage message) {
        try {
            //处理消息....每个服务的消息处理机制不一样,,,
            ServiceLoader<MessageArrived> load = ServiceLoader.load(MessageArrived.class);
            Iterator<MessageArrived> messIterator = load.iterator();
            Boolean notFound = true;
            while (messIterator.hasNext()) {
                notFound = false;
                MessageArrived messageArrived = messIterator.next();
                // 处理消息
                messageArrived.processMessage(s, message); //
            }
//
//           List<MessageArrived> messageArriveds = SpringFactoriesLoader.loadFactories(MessageArrived.class, null);
//            for (MessageArrived messageArrived : messageArriveds) {
//                System.out.println("springfa:::::::::::::::");
//            }
            if (notFound) {
                log.error("找不到具体实现");
            }
        } catch (Exception e) {
            log.error("处理mqtt消息异常:" + e);
        }
    }


    /**
     * 接收到消息调用令牌中调用
     *
     * @param iMqttDeliveryToken
     */
    @Override
    public void deliveryComplete(IMqttDeliveryToken iMqttDeliveryToken) {
        log.debug("deliveryComplete---------" + Arrays.toString(topic));
    }
}
